package com.zhongkai.app;


import com.zhongkai.MessageTool;
import org.apache.log4j.Logger;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RessageListener {

    private static final Logger logger = Logger.getLogger(RessageListener.class);

 //   public static  final ObjectMapper mapper = new ObjectMapper(); //解析json

    public  DefaultMQPushConsumer defaultMQPushConsumer = null;


    @Autowired
    private ListenerConcurrentlySky listenerConcurrentlySky;

    public  void init2() {
        defaultMQPushConsumer = new DefaultMQPushConsumer("myCat");
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        defaultMQPushConsumer.setInstanceName("myCat"); //不同的实例设置不同的name  否则无法同时存在
        try {
         //   defaultMQPushConsumer.subscribe(MessageTool.BIG_SKY, "*");
            defaultMQPushConsumer.subscribe(MessageTool.UNIT_SKY, "*");
            //    defaultMQPushConsumer.subscribe("ROCKETMQ_TOPIC", "*");
        } catch (MQClientException e) {
            System.out.println("失败MQClientException");
            e.printStackTrace();
        }
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING); //分布式模式 此处可做分布式处理
        defaultMQPushConsumer.registerMessageListener(listenerConcurrentlySky);

/*        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {


            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {



        });*/


        try {
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

}
